package com.matrix.draft.kafka.day2;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

/**
 * 指定分区进行消费
 */
public class MQTTConsumeMessageAssignPartition {

    public static void main(String[] args) {

        // 0. 配置
        Properties consumerProperties = new Properties();
        // 指定kafka服务器
        consumerProperties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "MatrixEngine:9092");
        // key的反序列化类
        consumerProperties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        // value的反序列化类
        consumerProperties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        // 消费者组id
        consumerProperties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "c1");

        // 1. 创建消费者组
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumerProperties);

        // 2. 订阅
        consumer.assign(Arrays.asList(new TopicPartition("first", 0)));

        // 3. 批次的拉数据
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(3));
            for (ConsumerRecord<String, String> record : records) {
                System.err.println(record);
            }
        }
    }

}
